package org.example.com.atguigu.day05;

import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SaveMode;
import org.apache.spark.sql.SparkSession;

import java.util.Properties;

public class WriterToFile {
    public static void main(String[] args) {
        // 从mysql中读数据
        SparkSession spark = SparkSession.builder().master("local[4]").appName("test").getOrCreate();
        String url = "jdbc:mysql://hadoop102:3306/gmall?useUnicode=true&characterEncoding=UTF-8";
        String tableName = "user_info";
        Properties props = new Properties();
        props.setProperty("user", "root");
        props.setProperty("password", "000000");
        Dataset<Row> ds = spark.read().jdbc(url, tableName, props);

        // 保存为Json
        ds.write().json("output/json");  // 目前一个分区,最终可以看到生成了一个json文件
        // ↑ 上面执行过,再执行一次,会报错:文件已存在
        // 可以修改为
        ds.write().mode(SaveMode.Append).json("output/json");  // 追加
        ds.write().mode(SaveMode.ErrorIfExists).json("output/json"); // 默认
        ds.write().mode(SaveMode.Ignore).json("output/json");  // 忽略
        ds.write().mode(SaveMode.Overwrite).json("output/json");  // 覆盖

        // 保存为文本 [保存为文本要求数据集只能有一个列]
        ds.selectExpr("login_name", "name").write().text("output.text");

        // 保存为csv
        // sep: 指定字段之间的分隔符
        // header: 将列名作为文件第一行
        // nullValue: 将null作为哪种字符串保存
        ds.write().mode(SaveMode.Overwrite).option("sep", "_").option("header", "true").csv("output/csv");
    }
}
